#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <algorithm>
#include <assert.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <iostream>
#include <vector>
#include <sys/time.h>
#include <string.h>
#if __GNUC_PREREQ(4,3)
#else
#include <byteswap.h>
#endif
#include "lf_vector.h"

#define MAX_LINE_LENGTH 16

struct Index
{
  int64_t intv[2];
  //char *orig;
  //int64_t psize;
  inline bool operator< (const Index &other) const
  {
    bool bret = false;
    int64_t cmp_ret = intv[0] - other.intv[0];
    if (0 == cmp_ret)
    {
      cmp_ret = intv[1] - other.intv[1];
    }
    bret = (0 > cmp_ret);
    return bret;
  };
};

struct MapInfo
{
  std::vector<Index> index;
  int64_t pos;
};

struct ReduceInfo
{
  volatile int64_t map_num;
  volatile int64_t map_done;
  lockfree::vector<Index> map_out;
  char *reduce_out;
  int64_t out_pos;
};

struct Partition
{
  void init(char *buffer, const int64_t size, const int64_t range_num)
  {
    std::vector<Index> array;
    srand(0);
    for (int64_t i = 1; i < range_num * 100; i++)
    {
      int64_t pos = rand() % size;
      while (pos < size)
      {
        if (buffer[pos] == 0x0d)
        {
          pos += 2;
          break;
        }
        if (buffer[pos] == 0x0a)
        {
          pos += 1;
          break;
        }
        pos++;
      }
      if (pos >= size)
      {
        i--;
        continue;
      }
      int64_t start_pos = pos;
      Index index;
      while (pos < size)
      {
        if (buffer[pos] == 0x0d)
        {
          int64_t psize = pos - start_pos;

          index.intv[0] = 0;
          index.intv[1] = 0;
          memcpy(index.intv, &buffer[start_pos], psize);
#if __GNUC_PREREQ(4,3)
          index.intv[0] = __builtin_bswap64(index.intv[0]);
          index.intv[1] = __builtin_bswap64(index.intv[1]);
#else
          index.intv[0] = __bswap_64(index.intv[0]);
          index.intv[1] = __bswap_64(index.intv[1]);
#endif
          array.push_back(index);

          pos += 2;
          break;
        }
        pos++;
      }
    }
    std::sort(array.begin(), array.end());
    for (int64_t i = 1; i < range_num; i++)
    {
      array_.push_back(array[i * 99]);
    }
  }
  inline int64_t get_part(Index &index) const
  {
    std::vector<Index>::const_iterator ret_iter;
    ret_iter = upper_bound(array_.begin(), array_.end(), index);
    return (ret_iter - array_.begin());
  }
  std::vector<Index> array_;
};

struct MergeIndex
{
  Index index;
  int64_t way;
};

inline int64_t tv_to_microseconds(const timeval & tp)
{
  return (((int64_t) tp.tv_sec) * 1000000 + (int64_t) tp.tv_usec);
}

inline int64_t get_cur_microseconds_time(void)
{
  struct timeval tp;
  gettimeofday(&tp, NULL);
  return tv_to_microseconds(tp);
}


void map_func(bool is_first, char *buffer, char *buffer_end, const int64_t size,
              const int64_t map_index, std::vector<ReduceInfo> &reduce, Partition &partition)
{
  int64_t size2read = size + MAX_LINE_LENGTH * 2;
  int64_t read_ret = std::min(size2read, (int64_t)(buffer_end - buffer));

  int64_t buf_pos = 0;
  if (!is_first)
  {
    while (buf_pos < read_ret)
    {
      if (buffer[buf_pos] == 0x0d)
      {
        buf_pos += 2;
        break;
      }
      if (buffer[buf_pos] == 0x0a)
      {
        buf_pos += 1;
        break;
      }
      buf_pos++;
    }
  }

  Index index;
  while (buf_pos < read_ret)
  {
    int64_t start_pos = buf_pos;
    while (buf_pos < read_ret)
    {
      if (buffer[buf_pos] == 0x0d)
      {
        int64_t psize = buf_pos - start_pos;
        //assert(0 < psize);
        //assert(16 >= psize);

        index.intv[0] = 0;
        index.intv[1] = 0;
        memcpy(index.intv, &buffer[start_pos], psize);
#if __GNUC_PREREQ(4,3)
        index.intv[0] = __builtin_bswap64(index.intv[0]);
        index.intv[1] = __builtin_bswap64(index.intv[1]);
#else
        index.intv[0] = __bswap_64(index.intv[0]);
        index.intv[1] = __bswap_64(index.intv[1]);
#endif
        //index.orig = &(buffer[start_pos]);
        //index.psize = psize + 2;

        buf_pos += 2;
        
        int64_t part = partition.get_part(index);
        reduce[part].map_out.push_back(index);
        break;
      }
      buf_pos++;
    }
    
    if (buf_pos > size)
    {
      break;
    }
  }

  std::vector<ReduceInfo>::iterator riter;
  for (riter = reduce.begin(); riter != reduce.end(); riter++)
  {
    __sync_add_and_fetch(&(riter->map_done), 1);
  }
}

inline void reduce_func(lockfree::vector<Index> &map_out, char *reduce_out, int64_t &out_pos)
{
  map_out.sort();
  char enter[2] = {0x0d, 0x0a};
  for (int64_t i = 0; i < map_out.size(); i++)
  {
    Index &index = map_out[i];
#if __GNUC_PREREQ(4,3)
    index.intv[0] = __builtin_bswap64(index.intv[0]);
    index.intv[1] = __builtin_bswap64(index.intv[1]);
#else
    index.intv[0] = __bswap_64(index.intv[0]);
    index.intv[1] = __bswap_64(index.intv[1]);
#endif
    char *orig = (char*)(index.intv);
    int64_t l = strnlen(orig, 16);
    memcpy(reduce_out + out_pos, orig, l);
    memcpy(reduce_out + out_pos + l, enter, sizeof(enter));
    //memcpy(reduce_out + out_pos, index.orig, index.psize);
    out_pos += (l + sizeof(enter));
  }
}

struct ThreadParam
{
  bool is_first;
  char *buffer;
  char *buffer_end;
  int64_t size;
  int64_t thread_index;
  std::vector<ReduceInfo> *reduce;
  Partition *partition;
};

void *thread_func(void *data)
{
  ThreadParam *tp = (ThreadParam*)data;
  int64_t timeu = get_cur_microseconds_time();
  map_func(tp->is_first, tp->buffer, tp->buffer_end, tp->size,
          tp->thread_index, *(tp->reduce), *(tp->partition));
  //fprintf(stderr, "map timeu=%ld\n", get_cur_microseconds_time() - timeu);

  while ((*(tp->reduce))[tp->thread_index].map_num != (*(tp->reduce))[tp->thread_index].map_done);

  timeu = get_cur_microseconds_time();
  reduce_func((*(tp->reduce))[tp->thread_index].map_out,
              (*(tp->reduce))[tp->thread_index].reduce_out,
              (*(tp->reduce))[tp->thread_index].out_pos);
  //fprintf(stderr, "reduce timeu=%ld\n", get_cur_microseconds_time() - timeu);

  return NULL;
}

inline void master_func(int ifd, int ofd, const int64_t total_size, const int64_t thread_num)
{
  int64_t timeu = get_cur_microseconds_time();

  ThreadParam *tp = new ThreadParam[thread_num];
  assert(NULL != tp);
  pthread_t *td = new pthread_t[thread_num];
  assert(NULL != td);
  char *buffer = new char[total_size];
  assert(NULL != buffer);

  int64_t read_ret = read(ifd, buffer, total_size);
  assert(total_size == read_ret);
  fprintf(stderr, "read timeu=%ld\n", get_cur_microseconds_time() - timeu);

  std::vector<ReduceInfo> &reduce = *(new std::vector<ReduceInfo>());
  Partition &partition = *(new Partition());;

  //int64_t addv = (thread_num - 1) * 2 * total_size / (3 * (thread_num - 1) * thread_num);
  //int64_t base = 2 * total_size / (3 * (thread_num - 1) * thread_num);
  int64_t part_size = total_size / thread_num;

  partition.init(buffer, total_size, thread_num);
  fprintf(stderr, "build partition timeu=%ld\n", get_cur_microseconds_time() - timeu);

  reduce.resize(thread_num);
  std::vector<ReduceInfo>::iterator riter;
  for (riter = reduce.begin(); riter != reduce.end(); riter++)
  {
    riter->map_num = thread_num;
    riter->map_done = 0;
    riter->map_out.resize(300000); // TODO read total lines
    riter->reduce_out = (char*)malloc(total_size);
    riter->out_pos = 0;
  }
  fprintf(stderr, "init reduce timeu=%ld\n", get_cur_microseconds_time() - timeu);

  for (int64_t i = 0; i < thread_num; i++)
  //for (int64_t i = thread_num - 1; i >= 0; i--)
  {
    tp[i].is_first = (0 == i);
    tp[i].buffer = buffer + i * part_size;
    //tp[i].buffer = buffer + i * base + (i * (i - 1) / 2) * addv;
    tp[i].buffer_end = buffer + total_size;
    if (i == (thread_num - 1))
    {
      tp[i].size = total_size - (i * part_size);
      //tp[i].size = total_size - (i * base + (i * (i - 1) / 2) * addv);
    }
    else
    {
      tp[i].size = part_size;
      //tp[i].size = base + i * addv;
    }
    tp[i].thread_index = i;
    tp[i].reduce = &reduce;
    tp[i].partition = &partition;
  }
  for (int64_t i = thread_num - 1; i >= 0; i--)
  {
    pthread_create(&(td[i]), NULL, thread_func, &(tp[i]));
  }
  for (int64_t i = 0; i < thread_num; i++)
  {
    pthread_join(td[i], NULL);
  }
  fprintf(stderr, "sort timeu=%ld\n", get_cur_microseconds_time() - timeu);

  for (riter = reduce.begin(); riter != reduce.end(); riter++)
  {
    int64_t write_ret = write(ofd, riter->reduce_out, riter->out_pos);
    assert(riter->out_pos == write_ret);
  }
  fprintf(stderr, "write timeu=%ld\n", get_cur_microseconds_time() - timeu);

  delete[] buffer;
  delete[] td;
  delete[] tp;
}

int main(int argc, char **argv)
{
  char *ifname = argv[1];
  char *ofname = argv[2];
  int64_t thread_num = atoll(argv[3]);
  int ifd = open(ifname, O_RDONLY);
  assert(-1 != ifd);
  int ofd = open(ofname, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
  assert(-1 != ofd);
  struct stat st;
  fstat(ifd, &st);

  int64_t timeu = get_cur_microseconds_time();
  master_func(ifd, ofd, st.st_size, thread_num);
  fprintf(stderr, "total timeu=%ld\n", get_cur_microseconds_time() - timeu);
}

